/*
 *  ------------------------------------------------------------------
 *  Copyright © 2017 Hangzhou DtDream Technology Co.,Lt d. All rights reserved.
 *  ------------------------------------------------------------------
 *              Product: EMR
 *    Module Name: DataTrain
 *      Date Created: 2017-10-23
 *         Description:
 *  ------------------------------------------------------------------
 *  Modification History
 *  DATE            Name           Description
 *  ------------------------------------------------------------------
 *  2017-10-23     XZ 209
 *  ------------------------------------------------------------------
 */

package org.xukai.remoting.sdk.rpc.transfer;



import com.dtdream.emr.transmission.exception.TransmissionException;
import com.dtdream.emr.transmission.io.Constant;
import com.dtdream.emr.transmission.io.Request;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.UUID;
import org.xukai.remoting.sdk.rpc.job.DownloadJob;
import org.xukai.remoting.sdk.rpc.job.Job;
import org.xukai.remoting.sdk.rpc.job.UploadJob;
import org.xukai.remoting.sdk.rpc.reader.Reader;
import org.xukai.remoting.sdk.rpc.writer.Writer;

public class ClientHandler extends ChannelInboundHandlerAdapter {
    private static Logger LOG = LoggerFactory.getLogger(ClientHandler.class);

    public ClientHandler() {
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws InterruptedException, TransmissionException {
        Request request = (Request) msg;
        if (request.getJobId() == null) {
            throw new TransmissionException(TransmissionException.Error.EXCEPTION_JOB_INVALID_ID);
        }

        UUID jobId = UUID.fromString(request.getJobId());
        int id = request.getId();
        int type = request.getMessageType();
        Job job = Job.getJob(jobId);

        switch (type) {
            case Constant.RECORDS:
                if (job instanceof DownloadJob) {
                    DownloadJob downloadJob = (DownloadJob) job;
                    Reader reader = downloadJob.getReader(id);
                    reader.buffer(request.getRecord());
                }
                break;
            case Constant.READ_FINISHED:
                if (job instanceof DownloadJob) {
                    DownloadJob downloadJob = (DownloadJob) job;
                    Reader reader = downloadJob.getReader(id);
                    reader.finish();
                }
                break;
            case Constant.ERROR_MESSAGE:
                TransmissionException exception = new TransmissionException(TransmissionException.Error.valueOf(request.getError()));
                if (job instanceof DownloadJob) {
                    readErrorProcess(job, id, exception);
                } else if (job instanceof UploadJob) {
                    writeErrorProcess(job, id, exception);
                }
                break;
            default:
                break;
        }
    }

    private void readErrorProcess(Job job, int readerId, TransmissionException exception) {
        DownloadJob downloadJob = (DownloadJob) job;
        Reader reader = downloadJob.getReader(readerId);
        if (null != reader) {
            reader.setError(exception);
            reader.finish();
        }
    }

    private void writeErrorProcess(Job job, int writerId, TransmissionException exception) {
        UploadJob uploadJob = (UploadJob) job;
        Writer writer = uploadJob.getWriter(writerId);
        if (null != writer) {
            writer.setError(exception);
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        LOG.error("ExceptionCaught: " + cause.toString(), cause);
        ctx.close();
    }
}
